/**
 *     This file is part of cesm-ws.
 *
 *     cesm-ws, Web Service for Community Earth System Model
 *
 *     Copyright (c) 2010, Purdue University and/or affilicates or thrid-party contributors as
 *     indicated by the @author tags or express copyright attribution statements
 *     applied by the authors. All third-party contributions are distributed under
 *     license by Purdue University.
 *
 *     cesm-ws is free software: you can redistribute it and/or modify
 *     it under the terms of the GNU General Public License as published by
 *     the Free Software Foundation, either version 3 of the License, or
 *     (at your option) any later version.
 *
 *     This program is distributed in the hope that it will be useful,
 *     but WITHOUT ANY WARRANTY; without even the implied warranty of
 *     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *     GNU General Public License for more details.
 *
 *     You should have received a copy of the GNU Lesser General Public License
 *     along with this distribution.  If not, see <http://www.gnu.org/licenses/>.
 */

package edu.purdue.rcac.grid.submit.globus;

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;

import org.apache.log4j.Logger;
import org.globus.ftp.exception.FTPException;
import org.globus.gram.Gram;
import org.globus.gram.GramException;
import org.globus.gram.GramJob;
import org.globus.gsi.GlobusCredential;
import org.globus.gsi.GlobusCredentialException;
import org.globus.gsi.gssapi.GlobusGSSCredentialImpl;
import org.globus.io.streams.GridFTPInputStream;
import org.globus.io.streams.GridFTPOutputStream;
import org.globus.rsl.NameOpValue;
import org.globus.rsl.ParseException;
import org.globus.rsl.RSLParser;
import org.globus.rsl.RslNode;
import org.ietf.jgss.GSSCredential;
import org.ietf.jgss.GSSException;

/**
 * Util class for Globus job submission
 * 
 * @author Han Zhang
 * @author Madhavan Lakshminarayanan
 * 
 */
public class GlobusSubmitter extends GlobusTranslator {

	private static String classDir;
	private static String proxyFile;
	private static String proxyScript;

	static {
		// Static initializer to retrieve path of proxy file
		String path = GlobusSubmitter.class.getProtectionDomain()
				.getCodeSource().getLocation().getPath();
		// XXX: what happens if path separated by windows style '\' ?
		path = path.substring(path.indexOf('/'));

		int pos = path.indexOf(".jar!");
		if (pos > -1)
			classDir = path.substring(0, path.lastIndexOf('/', pos) + 1);
		else
			classDir = path.substring(0, path.lastIndexOf('/') + 1);
		proxyFile = classDir + "x509up";
		proxyScript = classDir + "proxy-init.sh";
	}

	/**
	 * resource specification language (RSL) string.
	 * 
	 * @see {@link http://www.globus.org/toolkit/docs/2.4/gram/rsl_spec1.html
	 *      RSL 1.0}
	 */
	private String rsl;

	/**
	 * Globus stdout string
	 */
	private String out;

	/**
	 * Globus stderr string
	 */
	private String err;

	/**
	 * GramJob object that stores current job
	 */
	private GramJob job;

	/**
	 * Credential instance
	 */
	private GlobusGSSCredentialImpl cred;

	/**
	 * Log4j logger
	 */
	private static final Logger logger = Logger.getLogger(GlobusSubmitter.class
			.getName());

	/**
	 * Constructor
	 * 
	 * @throws GSSException
	 * @throws GlobusCredentialException
	 */
	public GlobusSubmitter() throws GlobusCredentialException, GSSException {
		rsl = "";
		out = "";
		err = "";
		job = null;
		cred = generateCredential();
	}

	/**
	 * Set rsl with given string. The input string will be parsed first, if
	 * failed, ParseException will be thrown.
	 * 
	 * @param rsl
	 * @throws ParseException
	 */
	public void setRsl(String rsl) throws ParseException {
		// Parse RSL for any grammar error
		RSLParser.parse(rsl);
		logger.debug("RSL: " + rsl);
		this.rsl = rsl;
	}

	public void setCredential(GlobusGSSCredentialImpl cred) {
		this.cred = cred;
	}

	/**
	 * @return rsl string
	 */
	public String getRsl() {
		return rsl;
	}

	/**
	 * @return current {@link GramJob} instance
	 */
	public GramJob getJob() {
		return job;
	}

	/**
	 * Set current job
	 * 
	 * @param job
	 *            GramJob instance
	 */
	public void setJob(GramJob job) {
		this.job = job;
	}

	/**
	 * Parse current rsl and get the value of specified attribute
	 * 
	 * @param attribute
	 *            RSLAttribute enum type object
	 * @return value of attribute
	 * @throws ParseException
	 * @throws NullPointerException
	 */
	private String getRSLAttributeValue(RSLAttribute attribute)
			throws ParseException, NullPointerException {
		if (rsl == null) {
			throw new NullPointerException("Rsl is empty in  instance");
		}
		NameOpValue value = RSLParser.parse(rsl).getParam(attribute.name);
		if (value == null) {
			throw new NullPointerException("Given rsl does not include '"
					+ attribute.name + "' attribute");
		}
		return value.getValues().get(0).toString();
	}

	/**
	 * @return globus output file path if specified in rsl
	 * @throws ParseException
	 * @throws NullPointerException
	 */
	public String getOutFileName() throws ParseException, NullPointerException {
		return this.getRSLAttributeValue(RSLAttribute.STDOUT);
	}

	/**
	 * @return globus error file path if specified in rsl
	 * @throws ParseException
	 * @throws NullPointerException
	 */
	public String getErrFileName() throws ParseException, NullPointerException {
		return this.getRSLAttributeValue(RSLAttribute.STDERR);
	}

	/**
	 * @return globus output file content if path specified in rsl
	 * @throws NullPointerException
	 * @throws IOException
	 * @throws FTPException
	 * @throws ParseException
	 */
	public String getOutContent() throws NullPointerException, IOException,
			FTPException, ParseException {
		this.out = getFileContent(Server.STEELE.dataServerName,
				Server.STEELE.dataPort, getOutFileName());
		return this.out;
	}

	/**
	 * @return globus error file content if path specified in rsl
	 * @throws NullPointerException
	 * @throws IOException
	 * @throws FTPException
	 * @throws ParseException
	 */
	public String getErrContent() throws NullPointerException, IOException,
			FTPException, ParseException {
		this.err = getFileContent(Server.STEELE.dataServerName,
				Server.STEELE.dataPort, getErrFileName());
		return this.err;
	}

	/**
	 * Submits the job specified with given rsl. It also start an
	 * InteractiveJobListener to track job status change. This method terminates
	 * only when job status changed to either FAILED or DONE.
	 * 
	 * @param serverName
	 *            name of server
	 * @param rsl
	 *            RSL string
	 * @throws ParseException
	 * @throws GramException
	 * @throws GSSException
	 * @throws InterruptedException
	 */
	public void submit(final String serverName, String rsl)
			throws ParseException, GramException, GSSException,
			InterruptedException {
		long start = System.currentTimeMillis();

		this.setRsl(rsl);
		RslNode root = RSLParser.parse(rsl);
		String executable = root.getParam(RSLAttribute.EXECUTABLE.name)
				.getValues().get(0).toString();

		long curr = System.currentTimeMillis();
		logger.debug(String.format("parse RSL cost %d ms = %.3f s", curr
				- start, (curr - start) / 1000f));
		start = curr;

		logger.debug("executable: " + executable);

		job = new GramJob(cred, rsl);

		curr = System.currentTimeMillis();
		logger.info(String.format("create gram job cost %d ms = %.3f s", curr
				- start, (curr - start) / 1000f));
		start = curr;

		logger.info("start running");
		Thread listenerThread = new Thread(new Runnable() {

			public void run() {
				String status = null;

				int i = 0;
				do {
					try {
						Thread.sleep(500);
						logger.info((++i) / 2. + " seconds : "+job.getStatusAsString());
					} catch (InterruptedException e1) {
						e1.printStackTrace();
					}
					try {
						Gram.jobStatus(job);
					} catch (GramException e) {
						if (e.getErrorCode() == GramException.ERROR_CONTACTING_JOB_MANAGER) {
							if (status == null) {
								logger.warn("Error contacting job manager - could not get job status");
							} else {
								logger.warn("Error contacting job manager - assuming job is finished.");
								break;
							}
						} else {
							logger.error("Failed to get job status: "
									+ e.getMessage());
						}
					} catch (GSSException gsse) {
						gsse.printStackTrace();
					}

					status = job.getStatusAsString();
					logger.debug("Globus: Status = " + status);
				} while (!status.equals("DONE") && !status.equals("FAILED"));

				Gram.deactivateAllCallbackHandlers();
				logger.info(String.format(
						"Globus: Job submission costs %.1f seconds", i / 2.));
			}

		});

		job.request(serverName, false);
		logger.info("successfully run job request");

		listenerThread.start();
		listenerThread.join();

		curr = System.currentTimeMillis();
		logger.debug(String.format("jobThread.start cost %d ms = %.3f s", curr
				- start, (curr - start) / 1000f));
		start = curr;

		curr = System.currentTimeMillis();
		logger.debug(String.format("while loop cost %d ms = %.3f s", curr
				- start, (curr - start) / 1000f));
		start = curr;
	}

	/**
	 * Generate credential from proxy file in class path
	 * 
	 * @return credential instance if successful
	 * @throws Exception
	 */
	public static GlobusGSSCredentialImpl generateCredential()
			throws GlobusCredentialException, GSSException {
		try {
			logger.debug("Proxy filename is " + proxyFile);

			GlobusCredential cred = new GlobusCredential(proxyFile);
			GlobusGSSCredentialImpl gssCred = new GlobusGSSCredentialImpl(cred,
					GSSCredential.INITIATE_AND_ACCEPT);

			logger.debug("Created GSS credential");

			return gssCred;
		} catch (Exception ex) {
			logger.info("Globus: Creating new proxy certificate");
			try {
				Runtime.getRuntime()
						.exec(proxyScript, null, new File(classDir)).waitFor();
			} catch (Exception ex2) {
				// Renew script failed; do nothing, will fail anyway below
			}

			logger.debug("Retrying: Proxy filename is " + proxyFile);

			// If this fails again, another exception will be thrown and we
			// wont't catch it this time
			GlobusCredential cred = new GlobusCredential(proxyFile);
			GlobusGSSCredentialImpl gssCred = new GlobusGSSCredentialImpl(cred,
					GSSCredential.INITIATE_AND_ACCEPT);

			logger.debug("Created GSS credential");

			return gssCred;
		}
	}

	/**
	 * Submits a job with current or given new credential. If override is set to
	 * be true, new credential is generated with
	 * {@link #generateCredential(String)}. Otherwise, as long as credential is
	 * not null, old credential will be used.
	 * 
	 * @param serverName
	 *            name of server
	 * @param rsl
	 *            rsl string
	 * @param override
	 *            used to decide if credential regeneration is needed
	 * @throws GlobusCredentialException
	 * @throws GSSException
	 * @throws GramException
	 * @throws ParseException
	 * @throws InterruptedException
	 */
	public void submit(String serverName, String rsl, boolean override)
			throws GlobusCredentialException, GSSException, GramException,
			ParseException, InterruptedException {
		if (override || cred == null)
			cred = generateCredential();
		this.submit(serverName, rsl);
	}

	/**
	 * Read file content remotely, will fail if credential is invalid.
	 * 
	 * @param serverName
	 * @param serverPort
	 * @param path
	 *            path to target file
	 * @return file content as string
	 * @throws IOException
	 * @throws FTPException
	 */
	public String getFileContent(String serverName, int serverPort, String path)
			throws IOException {
		// FIXME: notice that this outFileName could contain a string which
		// starts with '$'
		long start = System.currentTimeMillis();
		String content = "";
		try {
			GridFTPInputStream gis = new GridFTPInputStream(cred, serverName,
					serverPort, path);
			BufferedReader reader = new BufferedReader(new InputStreamReader(
					gis));
			String line;
			while ((line = reader.readLine()) != null) {
				content += line + "\n";
			}
			reader.close();
		} catch (FTPException ftpe) {
			throw new IOException(ftpe);
		}
		long end = System.currentTimeMillis();
		logger.info(String.format("Globus: Read file '%s' costs %.2f seconds",
				path, (end - start) / 1000.));
		return content;
	}

	/**
	 * 
	 * @param content
	 * @param serverName
	 * @param serverPort
	 * @param path
	 * @param append
	 * @throws IOException
	 * @throws FTPException
	 */
	public void setFileContent(String content, String serverName,
			int serverPort, String path, boolean append) throws IOException,
			FTPException {
		long start = System.currentTimeMillis();
		logger.info(String.format("Globus: Changing file '%s' ...", path));
		GridFTPOutputStream gos = new GridFTPOutputStream(cred, serverName,
				serverPort, path, append);

		BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(gos));
		writer.write(content);
		writer.close();
		long end = System.currentTimeMillis();
		logger.info(String.format("Globus: Write file '%s' costs %.2f seconds",
				path, (end - start) / 1000.));
	}

	public GridFTPOutputStream getOutputStream(String toPath,
			String serverName, int serverPort, boolean append)
			throws IOException {
		try {
			logger.info(String.format("Globus: Getting output stream for '%s'",
					toPath));
			GridFTPOutputStream gos = new GridFTPOutputStream(cred, serverName,
					serverPort, toPath, append);
			return gos;
		} catch (FTPException ftpe) {
			throw new IOException(ftpe);
		}
	}

	public GridFTPInputStream getInputStream(String fromPath,
			String serverName, int serverPort) throws IOException {
		try {
			logger.info(String.format("Globus: Getting input stream for '%s'",
					fromPath));
			GridFTPInputStream gis = new GridFTPInputStream(cred, serverName,
					serverPort, fromPath);
			return gis;
		} catch (FTPException ftpe) {
			throw new IOException(ftpe);
		}
	}
}
